package com.atguigu.tingshu.search.mq;

import com.alibaba.fastjson.JSON;
import com.atguigu.tingshu.common.constant.KafkaConstant;
import com.atguigu.tingshu.search.service.SearchService;

;
import com.atguigu.tingshu.vo.album.TrackStatMqVo;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.util.StringUtils;

import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;

@Slf4j
@Configuration
public class SearchReceiver {

    //监听kafka消息的值,来判断是否上架
    @Autowired
    private SearchService searchService;

    @Autowired
    private RedisTemplate redisTemplate;

    //在这里获取编辑isopen的kafka消息
    @KafkaListener(topics = KafkaConstant.QUEUE_ALBUM_UPPER)
    public void upperAlbum(ConsumerRecord<String,String> record){

        String albumId = record.value();
        if(!StringUtils.isEmpty(albumId)){
            searchService.upperAlbum(Long.parseLong(albumId));
            log.info("专辑上架成功了....");

        }
    }

    @KafkaListener(topics = KafkaConstant.QUEUE_ALBUM_LOWER)
    public void lowerAlbum(ConsumerRecord<String,String> record){
        String albumId = record.value();
        if(!StringUtils.isEmpty(albumId)){
            searchService.lowerAlbum(Long.parseLong(albumId));
            log.info("专辑下架成功了....");
        }
    }


    @KafkaListener(topics = KafkaConstant.QUEUE_ALBUM_STAT_UPDATE )
    public void updateAlbumStat(ConsumerRecord<String , String> record) {

        String value = record.value();
        if(!StringUtils.isEmpty(value)) {

            TrackStatMqVo trackStatMqVo = JSON.parseObject(value, TrackStatMqVo.class);
            String businessNo = trackStatMqVo.getBusinessNo();
            String redisKey = "album:stat:idempotent:" + businessNo ;

            // setIfAbsent方法底层调用的是redis的setnx命令，setnx表示如果key不存在，那么此时设置数据；如果key存在数据设置失败
            Boolean setIfAbsent = redisTemplate.opsForValue().setIfAbsent(redisKey, businessNo, 1, TimeUnit.HOURS);
            if(setIfAbsent) {
                // 调用service中的方法完成播放量的统计信息修改
                searchService.albumStatUpdate(trackStatMqVo) ;
            }

        }
    }
}
